#include "ThreadPool.hpp"
#include "BaseLog.hpp"
#include "Defind.hpp"
#include <thread>

using namespace DBServer::Base;

void
ThreadPool::work() {
    // init thread
    for (int i = 0; i < THREAD_POOL_SIZE; i++) {
        pool.emplace_back([this]{
            while (!is_stop.load())
            {
                // get the task in the blocking queue instantly
                std::packaged_task<void(void)> task;
                {
                    /* if the blocking queue is empty, hang up to wait for task */
                    // lock the queue
                    std::unique_lock<std::mutex> lock(queue_get);
                    wait.wait(lock, [this] {
                        return is_stop.load() || !blocking_queue.empty();
                    });
                    if (blocking_queue.empty()) // wake up by accident, but the blocking queue is empty
                        continue;
                    task = std::move(blocking_queue.front());
                    blocking_queue.pop();
                }
                try
                {
                    // execute the task
                    task();
                }
                catch(const std::exception& e)
                {
                    error("%d : %s\n", std::this_thread::get_id(), e.what());
                    exit(1);
                }
            }
        });
    }
    info("ThreadPool start!\n");
}

void
ThreadPool::stop() {
    is_stop.store(true);
    // notify all working thread
    wait.notify_all();
    for (auto &t : pool)
    {
        if (t.joinable())
        {
            info("%d -> join!\n", t.get_id());
            t.join();
        }
    }
}

void     
ThreadPool::commit(std::function<void(void)> f) {
    // if the thread pool already stop or isn't begin to work
    if (is_stop.load()) {
        warn("what():commit fail,the threal pool is not in working.");
        return;
    }
    try {
        // package task
        auto task = std::make_shared<std::packaged_task<void(void)>>(f);
        // lock the queue to commit the task
        {
            std::unique_lock<std::mutex> lock(queue_mut);
            // node that here, a lambda expression and std::shared_ptr is required
            // to form a pseudo-closure to extend the file cycle of the task
            // catch the task by copy but not refrence because copy construct can
            // add the refence counting of the task
            blocking_queue.emplace([task]{
                (*task)();
            });
        }
    } catch (const std::exception& e) {
        error(e.what());
        return;
    }
    // wake up a thread to get the task
    wait.notify_one();
    return;
}